package com.ruoyi.mqtt.netty;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import com.ruoyi.mqtt.handler.MqttMessageHandler;

@Slf4j
public class MqttChannelHandler extends SimpleChannelInboundHandler<MqttMessage> {
    
    private final MqttMessageHandler messageHandler;
    
    public MqttChannelHandler(MqttMessageHandler messageHandler) {
        this.messageHandler = messageHandler;
    }
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) {
        if (msg.fixedHeader().messageType() == MqttMessageType.PUBLISH) {
            MqttPublishMessage publishMessage = (MqttPublishMessage) msg;
            byte[] payload = new byte[publishMessage.payload().readableBytes()];
            publishMessage.payload().readBytes(payload);
            
            Message<byte[]> message = MessageBuilder.withPayload(payload).build();
            messageHandler.handleMessage(message);
        }
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("MQTT客户端异常", cause);
        ctx.close();
    }
} 